package com.itheima.flinkmomo.flink;

import com.itheima.flinkmomo.pojo.MoMoCountBean;
import com.itheima.flinkmomo.utils.HttpClientUtils;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;


import java.util.Properties;

/**
 * @author lilulu
 * @date 2023/2/27 11:38
 */
public class MOMOFlink {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();

        //设置source组件，用于接收kafka中的消息数据
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "node1:9092,node2:9092,node3:9092");
        props.setProperty("group.id", "momo_g2");
        props.setProperty("enable.auto.commit", "true");
        props.setProperty("auto.commit.interval.ms", "1000");

        FlinkKafkaConsumer kafkaConsumer = new FlinkKafkaConsumer("MOMO_MSG", new SimpleStringSchema(), props);
        DataStreamSource dataStreamSource = executionEnvironment.addSource(kafkaConsumer);

        //3. 设置转换组件, 对消息数据, 进行实时处理操作
        // 需求1: 实时统计消息总条数
        totalMsgCount(dataStreamSource);
        // 需求二: 实时统计各个地区发送的消息总量
//        totalMsgSenderProvince(dataStreamSource);

        // 需求三: 实时统计各个地区接收的消息总量
//        totalMsgReceiverProvince(dataStreamSource);

        // 需求四: 实时统计各个用户发送的消息总量
        totalMsgSenderUser(dataStreamSource);

        // 需求五: 实时统计各个用户接收的消息总量
        totalMsgReceiverUser(dataStreamSource);

        //提交运行
        executionEnvironment.execute();
    }

    private static void totalMsgReceiverUser(DataStreamSource dataStreamSource) {
        SingleOutputStreamOperator filterOperator = dataStreamSource.filter(new FilterFunction<String>() {
            public boolean filter(String msg) throws Exception {
                if (msg != null && !"".equals(msg) && msg.split("\001").length == 20) {
                    return true;
                }
                return false;
            }
        });
        // 3.2:  对数据的进行转换操作: 将消息中发件人的姓名获取出来,将发件人姓名作为 key, value放置为
        SingleOutputStreamOperator mapOperator = filterOperator.map(new MapFunction<String, Tuple2<String, Long>>() {

            public Tuple2<String, Long> map(String s) throws Exception {
                String[] split = s.split("\001");
                String name = split[9];
                return new Tuple2<String, Long>(name, 1L);
            }
        });
        //3.3 分组统计即可
        SingleOutputStreamOperator sumOperator = mapOperator.keyBy(0).sum(1);
        //3.4 将 Tuple2<String, Long> 转换为  MoMoCountBean对象
        SingleOutputStreamOperator operator = sumOperator.map(new MapFunction<Tuple2<String, Long>, MoMoCountBean>() {

            public MoMoCountBean map(Tuple2<String, Long> stringLongTuple2) throws Exception {
                String userName = stringLongTuple2.f0;
                Long count = stringLongTuple2.f1;
                MoMoCountBean moMoCountBean = new MoMoCountBean();
                moMoCountBean.setMoMoUsername(userName);
                moMoCountBean.setMoMo_MsgCount(count);
                return moMoCountBean;
            }
        });
        operator.addSink(new MysqlSink("5"));
    }

    private static void totalMsgSenderUser(DataStreamSource dataStreamSource) {
        SingleOutputStreamOperator filterOperator = dataStreamSource.filter(new FilterFunction<String>() {
            public boolean filter(String msg) throws Exception {
                if (msg != null && !"".equals(msg) && msg.split("\001").length == 20) {
                    return true;
                }
                return false;
            }
        });
        // 3.2:  对数据的进行转换操作: 将消息中发件人的姓名获取出来,将发件人姓名作为 key, value放置为
        SingleOutputStreamOperator mapOperator = filterOperator.map(new MapFunction<String, Tuple2<String, Long>>() {

            public Tuple2<String, Long> map(String s) throws Exception {
                String[] split = s.split("\001");
                String name = split[1];
                return new Tuple2<String, Long>(name, 1L);
            }
        });
        //3.3 分组统计即可
        SingleOutputStreamOperator sumOperator = mapOperator.keyBy(0).sum(1);
        //3.4 将 Tuple2<String, Long> 转换为  MoMoCountBean对象
        SingleOutputStreamOperator operator = sumOperator.map(new MapFunction<Tuple2<String, Long>, MoMoCountBean>() {

            public MoMoCountBean map(Tuple2<String, Long> stringLongTuple2) throws Exception {
                String userName = stringLongTuple2.f0;
                Long count = stringLongTuple2.f1;
                MoMoCountBean moMoCountBean = new MoMoCountBean();
                moMoCountBean.setMoMoUsername(userName);
                moMoCountBean.setMoMo_MsgCount(count);
                return moMoCountBean;
            }
        });
        operator.addSink(new MysqlSink("4"));
    }

    private static void totalMsgReceiverProvince(DataStreamSource dataStreamSource) {
        SingleOutputStreamOperator filterOperator = dataStreamSource.filter(new FilterFunction<String>() {
            public boolean filter(String msg) throws Exception {
                if (msg != null && !"".equals(msg) && msg.split("\001").length == 20) {
                    return true;
                }
                return false;
            }
        });
        SingleOutputStreamOperator mapOperator = filterOperator.map(new MapFunction<String, Tuple2<String, Long>>() {

            public Tuple2<String, Long> map(String s) throws Exception {
                String[] split = s.split("\001");
                String[] split1 = split[15].split(",");
                String lat = split1[0];
                String lng = split1[1];
                String byLatAndLng = HttpClientUtils.findByLatAndLng(lng, lat);
                return new Tuple2<String, Long>(byLatAndLng, 1L);
            }
        });
        //3.3 分组统计即可
        SingleOutputStreamOperator sumOperator = mapOperator.keyBy(0).sum(1);
        //3.4 将 Tuple2<String, Long> 转换为  MoMoCountBean对象
        SingleOutputStreamOperator operator = sumOperator.map(new MapFunction<Tuple2<String, Long>, MoMoCountBean>() {

            public MoMoCountBean map(Tuple2<String, Long> stringLongTuple2) throws Exception {
                String longAndLat = stringLongTuple2.f0;
                Long count = stringLongTuple2.f1;
                MoMoCountBean moMoCountBean = new MoMoCountBean();
                moMoCountBean.setMoMoProvince(longAndLat);
                moMoCountBean.setMoMo_MsgCount(count);
                return moMoCountBean;
            }
        });
        operator.addSink(new MysqlSink("3"));
    }

    private static void totalMsgSenderProvince(DataStreamSource source) {
        SingleOutputStreamOperator filterOperator = source.filter(new FilterFunction<String>() {
            public boolean filter(String msg) throws Exception {
                if (msg != null && !"".equals(msg) && msg.split("\001").length == 20) {
                    return true;
                }
                return false;
            }
        });
        SingleOutputStreamOperator mapOperator = filterOperator.map(new MapFunction<String, Tuple2<String, Long>>() {

            public Tuple2<String, Long> map(String s) throws Exception {
                String[] split = s.split("\001");
                String[] split1 = split[8].split(",");
                String lat = split1[0];
                String lng = split1[1];
                String byLatAndLng = HttpClientUtils.findByLatAndLng(lng, lat);
                return new Tuple2<String, Long>(byLatAndLng, 1L);
            }
        });
        //3.3 分组统计即可
        SingleOutputStreamOperator sumOperator = mapOperator.keyBy(0).sum(1);
        //3.4 将 Tuple2<String, Long> 转换为  MoMoCountBean对象
        SingleOutputStreamOperator operator = sumOperator.map(new MapFunction<Tuple2<String, Long>, MoMoCountBean>() {

            public MoMoCountBean map(Tuple2<String, Long> stringLongTuple2) throws Exception {
                String longAndLat = stringLongTuple2.f0;
                Long count = stringLongTuple2.f1;
                MoMoCountBean moMoCountBean = new MoMoCountBean();
                moMoCountBean.setMoMoProvince(longAndLat);
                moMoCountBean.setMoMo_MsgCount(count);
                return moMoCountBean;
            }
        });
        operator.addSink(new MysqlSink("2"));

    }

    public static void totalMsgCount(DataStreamSource<String> source) {
        // 需求1: 实时统计消息总条数
        // 3.1: 设置过滤条件, 将不符合的数据过滤掉
        // 清洗要求:  数据不能为null 也不能为 空 同时 要求字段的数量必须为20
        SingleOutputStreamOperator filterOperator = source.filter(new FilterFunction<String>() {
            public boolean filter(String msg) throws Exception {
                if (msg != null && !"".equals(msg) && msg.split("\001").length == 20) {
                    return true;
                }
                return false;
            }
        });

        //3.2 转换操作:  将每一条数据转换为 1 ,进行累加求和计算
        SingleOutputStreamOperator mapOperator = filterOperator.map(new MapFunction<String, Tuple1<Long>>() {

            public Tuple1<Long> map(String msg) throws Exception {

                return new Tuple1<Long>(1L);
            }
        });
        //累加求和
        SingleOutputStreamOperator sumOperator = mapOperator.keyBy(0).sum(0);
        //3.3: 将 Tuple1<Long>  转换为  MoMoCountBean 对象
        SingleOutputStreamOperator operator = sumOperator.map(new MapFunction<Tuple1<Long>, MoMoCountBean>() {

            public MoMoCountBean map(Tuple1<Long> longTuple1) throws Exception {
                Long f0 = longTuple1.f0;
                MoMoCountBean moMoCountBean = new MoMoCountBean();
                moMoCountBean.setMoMoTotalCount(f0);
                return moMoCountBean;
            }
        });

        //4. 设置sink组件, 对数据进行输出操作:  将其保存到 MySQL中
        sumOperator.print();
        MysqlSink mysqlSink = new MysqlSink("1");
        operator.addSink(mysqlSink);
    }
    //需求2

}
